package com.hopu.shop;

import com.alibaba.fastjson.JSONObject;

import com.hopu.bean.shop.CateHotTop3;
import com.hopu.util.HBaseUtils;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.Optional;
import scala.Tuple2;

import java.io.IOException;
import java.util.List;

public class CateHotTop3Ana {

    public static void main(String[] args) {

        SparkConf conf = new SparkConf().setAppName("").setMaster("local");
        JavaSparkContext context = new JavaSparkContext(conf);
        JavaRDD<String> rdd = context.textFile("D:/word/user_session.log");

        JavaPairRDD<CateHotTop3, Integer> pairRDD = rdd.mapToPair(t -> {
            JSONObject json = (JSONObject) JSONObject.parse(t);
            String cid = json.getString("category_id");
            String cname = json.getString("category_name");
            String type = json.getString("event_type");

            return new Tuple2<>(new CateHotTop3(cid, cname, type), 1);
        });

        //过滤出不同操作类型的RDD
        JavaPairRDD<CateHotTop3, Integer> viewRDD = pairRDD.filter(t -> "view".equals(t._1.getType()));
        JavaPairRDD<CateHotTop3, Integer> cartRDD = pairRDD.filter(t -> "cart".equals(t._1.getType()));
        JavaPairRDD<CateHotTop3, Integer> parchaseRDD = pairRDD.filter(t -> "parchase".equals(t._1.getType()));

        //每个操作类型的RDD需要根据类目ID进行分组
        viewRDD = groupBySid(viewRDD);
        cartRDD = groupBySid(cartRDD);
        parchaseRDD = groupBySid(parchaseRDD);

        //合并三个不同操作类型的RDD
        JavaPairRDD<CateHotTop3, Integer> combine = combine(viewRDD, cartRDD, parchaseRDD);

        //排序
        JavaPairRDD<CateHotTop3, Integer> sortByKey = combine.sortByKey();

        //取出top3
        List<CateHotTop3> top3s = sortByKey.map(t -> t._1).take(3);
        top3s.forEach(System.out::println);

        //写入到HBase
        String[] columns = {"cid", "cname", "view", "cart", "parchase"};
        for (CateHotTop3 top3: top3s) {
            String[] values = {top3.getCid(), top3.getCname(), top3.getView()+"", top3.getCart()+"", top3.getParchase()+""};
            try {
                HBaseUtils.putOneRowToHbase("shopana", top3.getCid(), "top3hot", columns, values);
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }

    public static JavaPairRDD<CateHotTop3, Integer> combine(JavaPairRDD<CateHotTop3, Integer> viewRDD,
                               JavaPairRDD<CateHotTop3, Integer> cartRDD,
                               JavaPairRDD<CateHotTop3, Integer> parchaseRDD) {
        JavaPairRDD<CateHotTop3, Tuple2<Tuple2<Integer, Optional<Integer>>, Optional<Integer>>> result =
                viewRDD.leftOuterJoin(cartRDD).leftOuterJoin(parchaseRDD);

        JavaPairRDD<CateHotTop3, Integer> pairRDD = result.mapToPair(t -> {
            CateHotTop3 cateHotTop3 = t._1;
            int view = t._2._1._1;
            int cart = t._2._1._2.get();
            int parchase = t._2._2.get();
            cateHotTop3.setCart(cart);
            cateHotTop3.setParchase(parchase);
            return new Tuple2<>(cateHotTop3, view+cart+parchase);
        });
        return pairRDD;
    }

    public static JavaPairRDD<CateHotTop3, Integer> groupBySid(JavaPairRDD<CateHotTop3, Integer> rdd) {
        JavaPairRDD<CateHotTop3, Iterable<Integer>> group = rdd.groupByKey();

        JavaPairRDD<CateHotTop3, Integer> toPair = group.mapToPair(t -> {
            int count = 0;
            for (int v : t._2) {
                count += v;
            }
            if (t._1.getType().equals("view")) {
                t._1.setView(count);
            } else if (t._1.getType().equals("cart")) {
                t._1.setCart(count);
            } else {
                t._1.setParchase(count);
            }
            return new Tuple2<>(t._1, count);
        });
        return toPair;
    }
}
